package com.wyz.jdk8;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 *  jak CompletableFuture 演示类
 */
public class CompletableFutureDemo {

    public static final ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 4, 60L,
            TimeUnit.SECONDS, new LinkedBlockingQueue<>(1024), new ThreadFactory() {
        int count = 1;
        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, "custom-executor-"  + count++);
        }
    });
    /**
     *  主方法
     * @param args
     */
    public static void main(String[] args) throws InterruptedException {
        //================演示无参方法
//        runAsync();
//        thenApply();
//        thenApplyAsync();

//        thenAccept();
//        thenAcceptAsync();

//        completeExceptionally();

//        handle2();
//        cancel();
//        applyToEither();
//        acceptToEither();
//        runAfterBoth();
//        runAfterBoth2();
//        runAfterBothAsync();
//        thenAcceptBoth();
//        thenAcceptBothAsync();
//        thenCombine();
//        thenCompose();
//        anyOf();
        allOf();
//        allOfAsync();
        //================演示有参方法

//        thenApplyAsync(executor);
//        thenAcceptAsync(executor);

    }

    private static void allOfAsync() {
        List<Integer> list = Arrays.asList(1, 2, 3);
        List<CompletableFuture<Integer>> futures = list.stream().map(msg -> CompletableFuture.completedFuture(msg).thenApplyAsync(s -> {
            try {
                Thread.sleep(s*1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return s * 10;
        })).collect(Collectors.toList());
        CompletableFuture<Void> voidCompletableFuture = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])).whenComplete((t, u) -> {
            futures.forEach(fu -> {
                System.out.println(fu.getNow(null));
            });
        });

        voidCompletableFuture.join();

    }

    /**
     *  当等待所有的CompletableFuture 都执行完成  则创建一个完成的CompletableFuture.
     */
    private static void allOf() {
        List<Integer> list = Arrays.asList(1, 2, 3);
        List<CompletableFuture<Integer>> futures = list.stream().map(msg -> CompletableFuture.completedFuture(msg).thenApply(s -> {
            try {
                Thread.sleep(s*1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return s * 10;
        })).collect(Collectors.toList());
        CompletableFuture<Void> voidCompletableFuture = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])).whenComplete((t, u) -> {
            futures.forEach(fu -> {
                System.out.println(fu.getNow(null));
            });
        });

        voidCompletableFuture.join();

    }

    /**
     *  当多个CompletableFuture 中当任意一个CompletableFuture完成后， 则创建一个完成的CompletableFuture.
     */
    private static void anyOf() {
        StringBuilder sb = new StringBuilder();
        List<String> list = Arrays.asList("a", "b", "c");
        List<CompletableFuture<String>> futures = list.stream().map(msg -> CompletableFuture.completedFuture(msg).thenApplyAsync(s -> s.toUpperCase())).collect(Collectors.toList());
        CompletableFuture<Object> future = CompletableFuture.anyOf(futures.toArray(new CompletableFuture[futures.size()])).whenComplete((t, u) -> {
            //判断是否存在异常
            if (u == null) {
                // 判断是否完成大写转换
                if (((String) t).toUpperCase().equals(t)) {
                    sb.append(t);
                }

            }
        });
        System.out.println("结果： " + future.getNow(null));
    }

    /**
     * 将第一个CompletableFuture 的结果传递给 Function<? super T, ? extends CompletionStage<U>> 函数 最后返回的是Function的结果
     */
    private static void thenCompose() {
        String msg = "Hello";
        CompletableFuture<String> cf = CompletableFuture.completedFuture(msg).thenApply(t -> t.toUpperCase())
                .thenCompose(t-> CompletableFuture.completedFuture(msg).thenApply(s-> s.toLowerCase()).thenApply(s-> t + ": " + s ));
        System.out.println("结果： " + cf.join());

    }

    /**
     *  该方法是与 thenAcceptBoth 相似，不过区别是thenAcceptBoth 最后运行是消费者 没有返回结果
     *  而 thenCombine() 是有返回结果的
     *
     */
    private static void thenCombine() {
        String msg = "Hello";
        CompletableFuture<String> cf = CompletableFuture.completedFuture(msg).thenApply(t -> t.toUpperCase())
                .thenCombine(
                CompletableFuture.completedFuture(msg).thenApply(t -> t.toLowerCase()),
                (s1, s2) -> s1 +  ": " + s2);

        System.out.println(cf.getNow(null));
    }

    /**
     *  该方法等同于thenAcceptBoth
     */
    private static void thenAcceptBothAsync() {
        StringBuilder sb = new StringBuilder();
        String msg = "Hello";
        CompletableFuture<Void> cf = CompletableFuture.completedFuture(msg).thenApplyAsync(t -> t.toUpperCase(), CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)).thenAcceptBothAsync(
                CompletableFuture.completedFuture(msg).thenApplyAsync(t -> t.toLowerCase(), CompletableFuture.delayedExecutor(2, TimeUnit.SECONDS)),
                (s1, s2) -> sb.append(s1).append(": ").append(s2));
        cf.join();
        System.out.println("结果： " +sb);
    }


    /**
     *  通过测试得知 该thenAcceptBoth方法是将两个CompletableFuture执行的结果，进行合并处理
     */
    private static void thenAcceptBoth() {
        StringBuilder sb = new StringBuilder();
        String msg = "Hello";
        CompletableFuture.completedFuture(msg).thenApply(t-> t.toUpperCase()).thenAcceptBoth(
                CompletableFuture.completedFuture(msg).thenApply(t-> t.toLowerCase()),
                (s1,s2)-> sb.append(s1).append(": ").append(s2));
        System.out.println("结果： " +sb);
    }

    private static void runAfterBothAsync() {
        StringBuilder sb = new StringBuilder();
        String msg = "Hello";
        CompletableFuture cf = CompletableFuture.completedFuture(msg).thenApplyAsync(t->{
            try {
                Thread.sleep(1000);
                System.out.println("run first");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return t.toUpperCase();
        }).runAfterBothAsync( CompletableFuture.completedFuture(msg).thenApplyAsync(t->{
            try {
                Thread.sleep(10000);
                System.out.println("run second");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            String lowerCase = t.toLowerCase();

            return lowerCase;
        }),()-> {
            System.out.println("run 方法执行了");
            sb.append("done");
        });
        cf.join();
        System.out.println("结果：" + sb.toString());

    }

    /**
     *  该方法是两个CompletableFuture 都同步执行完了才会执行 Runnable
     */
    private static void runAfterBoth() {
        StringBuilder sb = new StringBuilder();

        String msg = "Hello";
        CompletableFuture cf = CompletableFuture.completedFuture(msg).thenApply(t->{
            try {
                Thread.sleep(1000);
                System.out.println("run first");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return t.toUpperCase();
        }).runAfterBoth( CompletableFuture.completedFuture(msg).thenApply(t->{
            try {
                Thread.sleep(1000);
                System.out.println("run second");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            String lowerCase = t.toLowerCase();

            return lowerCase;
        }),()-> sb.append("done"));

        System.out.println("结果：" + sb.toString());
    }

    /**
     *  该方法是测试两个CompletableFuture 使用异步执行，可以发现 runAfterBoth 这个方法不管是同步执行还是异步执行，是会阻塞等到两个CompletableFuture 方法都执行完才会执行
     *  所以这里使用异步处理效果跟同步处理方式一致
     */
    private static void runAfterBoth2() {
        StringBuilder sb = new StringBuilder();

        String msg = "Hello";
        CompletableFuture cf = CompletableFuture.completedFuture(msg).thenApplyAsync(t->{
            try {
                Thread.sleep(1000);
                System.out.println("run first");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return t.toUpperCase();
        }).runAfterBoth( CompletableFuture.completedFuture(msg).thenApplyAsync(t->{
            try {
                Thread.sleep(10000);
                System.out.println("run second");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            String lowerCase = t.toLowerCase();

            return lowerCase;
        }),()->{
            System.out.println("run 执行了");
            sb.append("done");
        });
        System.out.println(cf.join());
        System.out.println("结果：" + sb.toString());
    }


    /**
     *  该功能与applyToEither方法相似 只不过最后应用的是一个消费者
     */
    private static void acceptToEither() {
        StringBuilder sb = new StringBuilder();

        String msg = "Hello";
        CompletableFuture cf = CompletableFuture.completedFuture(msg).thenApplyAsync(t -> {
            try {
                Thread.sleep(1000);
                System.out.println("run first");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return t.toUpperCase();
        }).acceptEither( CompletableFuture.completedFuture(msg).thenApplyAsync(t ->  {
            try {
                Thread.sleep(1000);
                System.out.println("run second");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return t.toLowerCase();
        }),t-> {
            System.out.println("Consumer 执行了");
            sb.append(t).append(": acceptToEither");
        });

        cf.join();
        System.out.println("结果：" + sb.toString());
    }

    /**
     *  通过测试发现，当两个CompletionStage中谁先执行完成，用去谁的执行结果进行函数操作.
     */
    private static void applyToEither() {
        String msg = "Hello";
        CompletableFuture<String> cf = CompletableFuture.completedFuture(msg).thenApplyAsync(t ->  {
            try {
                Thread.sleep(1000);
                System.out.println("run first");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return t.toUpperCase();
        } );
        CompletableFuture<String> toEither = cf.applyToEither(CompletableFuture.completedFuture(msg).thenApplyAsync(t -> {
            try {
                Thread.sleep(8000);
                System.out.println("run second");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return t.toLowerCase();
        }), t -> t + " ：applyToEither");
        System.out.println(toEither.join());

    }

    /**
     *  通过同步和异步测试，发现同步执行执行取消操作会失败，异步取消会成功
     *  分析源码：
     *      通过分析源码发现， cancel(boolean mayInterruptIfRunning) 并不是通过中断线程来操作的，
     *      而是通过 new CancellationException()一个取消的异常来处理的，所以cancel方法中传入的布尔值没有用。
     *      执行cancel方法的执行过程：首先判断result == null 如果result不为null 则不会执行CancellationException异常来执行取消操作
     *      所以，同步串行在执行cancel时 result肯定不为null 所以同步串行下cancel方法将不起作用
     *
     */
    private static void cancel() {
        //同步测试
//        CompletableFuture<Object> future = CompletableFuture.completedFuture("message").thenApply(t -> {
        // 异步测试
        CompletableFuture<Object> future = CompletableFuture.completedFuture("message").thenApplyAsync(t -> {
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return t.toUpperCase();
        });
        //执行取消
        System.out.println("是否取消成功：" + future.cancel(false));
        System.out.println("是否取消成功：" + future.cancel(true));

        System.out.println(future.getNow(null));

    }

    /**
     *  通过测试和追踪源码可以发现，当CompletableFuture中封装了一个异常时，在获取结果的时候，会对我们自己抛出的异常进行异常类型转化，
     *  统一转化为CompletionException异常
     *
     */
    public static void completeExceptionally() {
        CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase,
                CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));
        cf.completeExceptionally(new RuntimeException("主动抛出异常！！"));
        System.out.println("是否存在异常  :"+ cf.isCompletedExceptionally());
        try {
//            cf.join();
            System.out.println(cf.getNow(null));
        } catch(CompletionException ex) {
            System.out.println("抛出异常=" + ex.getCause().getMessage());
        }

    }

    public static void handle() {
        //定义一个延迟线程池
        Executor executor = CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS);
        CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase,
                executor);
        /**
         * 处理异常 如果发生异常会执行该方法
         * 参数t 是cf 的执行结果
         * 参数u 是cf 中抛出的异常
         */
        CompletableFuture handle = cf.handle((t, u) -> {
            if (u != null) {
                return "有异常发生，返回fail";
            }
            return "没有发生异常，返回success";
        });
        cf.completeExceptionally(new RuntimeException("主动抛出异常！！"));
        System.out.println("是否存在异常  :"+ cf.isCompletedExceptionally());
        try {
            cf.join();
            System.out.println(cf.getNow(null));
        } catch(CompletionException ex) {
            System.out.println("抛出异常=" + ex.getCause().getMessage());
        }
        handle.join();
        System.out.println("异常处理结果="+ handle.getNow(""));

    }

    /**
     *  通过handle() 测试方法可以得知，通过handle方法可以获取一个新的CompletableFuture 而原来的cf因为有异常抛出，在获取结果的时候会抛出异常
     *  而通过handle() 返回的新的CompletableFuture对象，不管cf是否有异常发生都会被执行，所以对handle()方法进行了代码优化，优化后的代码如下：
     *
     */
    public static void handle2() {
        //定义一个延迟线程池
        Executor executor = CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS);
        CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase,
                executor);
        /**
         * 处理异常 这里从cf 中分离出一个处理发生异常的CompletableFuture
         * 参数t 是cf 的执行结果
         * 参数u 是cf 中抛出的异常
         */
        CompletableFuture handle = cf.handle((t, u) -> {
            if (u != null) {
                return "有异常发生，返回fail";
            }
            return t;
        });
//        cf.completeExceptionally(new RuntimeException("主动抛出异常！！"));

        //判断是否有异常
        handle.join();
        System.out.println("获取处理的结果" + handle.getNow("null"));

    }

    /**
     *   该方法是执行线程池异步执行方法。如果对上一步的结果进行异步消费的时候，可以使用该方法。
     * @param executor
     */
    private static void thenAcceptAsync(ThreadPoolExecutor executor) {
        CompletableFuture<Void> success = CompletableFuture.completedFuture("success").thenAcceptAsync(t->{
            System.out.println("当前线程名称：" + Thread.currentThread().getName());
            System.out.println("当前线程是否是守护线程：  " + Thread.currentThread().isDaemon());
            //将字符串转大写
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        },executor);
        System.out.println("结果："+ success.getNow(null));
        success.join();
    }

    /**
     *  通过调试可以发现 该方法是异步执行方法。如果对上一步的结果进行异步消费的时候，可以使用该方法。
     */
    private static void thenAcceptAsync() {
        CompletableFuture<Void> success = CompletableFuture.completedFuture("success").thenAcceptAsync(t->{
            System.out.println("当前线程名称：" + Thread.currentThread().getName());
            System.out.println("当前线程是否是守护线程：  " + Thread.currentThread().isDaemon());
            //将字符串转大写
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        System.out.println("结果："+ success.getNow(null));
        success.join();
    }

    /**
     *  thenAccept方法中是一个消费者，如果需要对上一步的结果进行处理，且不许要返回结果，则可以使用该方法，
     *  如果需要处理后有返回结果的，则适用于thenApply() 方法，
     *  thenAccept() 和 thenApply() 都是串行执行。
     */
    private static void thenAccept() {
        CompletableFuture<Void> success = CompletableFuture.completedFuture("success").thenAccept(t -> {
            System.out.println("当前线程名称：" + Thread.currentThread().getName());
            System.out.println("当前线程是否是守护线程：  " + Thread.currentThread().isDaemon());
            //将字符串转大写
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

        });
        System.out.println("结果："+ success.getNow(null));

    }

    /**
     *  该方法与thenApplyAsync() 方法功能一致，主要不同是，该方法使用的线程池是可以指定的，而thenApplyAsync()是使用默认的线程池
     * @param executor
     */
    private static void thenApplyAsync(ThreadPoolExecutor executor) {
        CompletableFuture<String> success = CompletableFuture.completedFuture("success").thenApplyAsync(t -> {
            System.out.println("当前线程名称：" + Thread.currentThread().getName());
            System.out.println("当前线程是否是守护线程：  " + Thread.currentThread().isDaemon());
            //将字符串转大写
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return t.toUpperCase();
        },executor);
        System.out.println("结果："+ success.getNow(null));


    }

    /**
     * 通过执行调试可以发现thenApplyAsync是异步执行，使用 ForkJoinPool.commonPool()线程池进行处理任务.该方法不会阻塞getNow方法
     */
    private static void thenApplyAsync() {
        CompletableFuture<String> success = CompletableFuture.completedFuture("success").thenApplyAsync(t -> {
            System.out.println("当前线程名称：" + Thread.currentThread().getName());
            System.out.println("当前线程是否是守护线程：  " + Thread.currentThread().isDaemon());
            //将字符串转大写
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return t.toUpperCase();
        });
        System.out.println("结果："+ success.getNow(null));
        success.join();
    }

    /**
     *   thenApply 是对上一步执行的结果进行处理。与上一步是串行处理
     *   通过修改睡眠的时长，可以发现thenApply 会阻塞getNow()，直到thenApply执行完才会返回结果。
     */
    private static void thenApply() {
        CompletableFuture<String> success = CompletableFuture.completedFuture("success").thenApply(t -> {
            System.out.println("当前线程是否是守护线程：  " + Thread.currentThread().isDaemon());
            //将字符串转大写
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return t.toUpperCase();
        });
        System.out.println("结果："+ success.getNow(null));

    }

    /**
     *  测试
     *  打印结果：
     *   是否执行完成： false
     *   该线程是否是守护线程true
     *   是否执行完成： true
     *
     *  通过测试 可以发现runAsync方法是异步开启线程执行的，该子线程是以守护线程进行执行的。
     *  查看源码发现 该方法默认使用ForkJoinPool线程池执行。
     */
    private static void runAsync() throws InterruptedException {

        CompletableFuture<Void> cf = CompletableFuture.runAsync(() -> {
            System.out.println("该线程是否是守护线程" + Thread.currentThread().isDaemon());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        System.out.println("是否执行完成： "+ cf.isDone());
        Thread.sleep(2000);
        System.out.println("是否执行完成： "+ cf.isDone());
    }


}